package com.atguigu.upp.app;

import com.atguigu.upp.bean.TagInfo;
import com.atguigu.upp.bean.TagValueTypeConstant;
import com.atguigu.upp.bean.TaskInfo;
import com.atguigu.upp.bean.TaskTagRule;
import com.atguigu.upp.service.MysqlDBService;
import com.atguigu.upp.service.UPPUtil;
import lombok.extern.log4j.Log4j;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.spark.api.java.function.ForeachFunction;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;
import java.util.List;
import java.util.stream.Collectors;

/**
 * Created by Smexy on 2023/1/3
 *
 任务一：
   ①读取每一个task的 task_sql(mysql)
   ②HQL语句要么使用 hive执行，要么使用sparksql
        选择sparksql :
            原因：
                    a) sparksql的运行效率高于hive
                    b) 在任务二中，需要使用一种语法 pivot,只有sparksql有
                    c) 计算任务应该是统一的。
                            事实统计型
                            规则型
                            机器学习型(目前不涉及，要留有余地)
   ③为每个计算的任务，都设计一张结果表。
            结果表的名字，一般以计算的标签的名字作为表名。
   ④将mysql中读取的 task_sql 进行处理(替换其中的占位符)，执行运算
            如果计算的是有四级标签的三级标签。
            把sql计算的结果映射为四级标签值。

 ----------------
  技术:
        mybatis: JDBC,负责读取mysql中的元数据
        sparksql : 执行hql

 ------------------
    用户画像平台和程序交互的方式。
        画像平台在调用程序时，会固定传入两个参数。
            参数1：  要计算的taskid
            参数2：  当天的业务日期

 */
@Log4j
public class SqlExecuteApp
{
    public static void main(String[] args) throws IOException {

        //接收参数
        String taskId = args[0];
        String doDate = args[1];

       /* String taskId = "3";
        String doDate = "2020-06-14";*/

        //查询当前这个taskId对应的元数据信息
        SqlSessionFactory sqlSessionFactory = UPPUtil.getSqlSessionFactoryByConfig("mysql_config.xml");
        MysqlDBService mysqlDBService = new MysqlDBService(sqlSessionFactory.openSession());
        TaskInfo taskInfo = mysqlDBService.getTaskById(taskId);
        TagInfo tagInfo = mysqlDBService.getTagByTaskId(taskId);
        List<TaskTagRule> rules = mysqlDBService.getRulesByTaskId(taskId);

        //获取SparkSession
        SparkSession sparkSession = UPPUtil.getSparkSession("SqlExecuteApp");

        //确定建表语句
        String tableSql = createTableSql(tagInfo);
        String insertSql = createInsertSql(tagInfo, doDate, rules, taskInfo);

       sparkSession.sql(tableSql);
        sparkSession.sql(insertSql);


    }

    /*
insert overwrite table upp220828.TAG_POPULATION_ATTRIBUTES_NATURE_GENDER partition (dt='2020-06-14')
select
    -- 当前计算的任务是有四级标签的三级标签，把sql查询的结果感觉映射规则，映射为四级标签名
    uid,
    case tagValue
        when 'U' then '未知'
        when 'M' then '男性'
        when 'F' then '女性'
    end tagValue
from (
         select id                                uid,

                `if`(isnull(gender), 'U', gender) tagValue
         from gmall.dim_user_zip
         where dt = '9999-12-31'
     ) tmp
     */
    private static String createInsertSql(TagInfo tagInfo,String doDate,List<TaskTagRule> rules,TaskInfo taskInfo){

        String template = " insert overwrite table %s.%s partition (dt='%s') " +
                          " select  uid, %s from ( %s )tmp";

        //填充占位符
        String db = UPPUtil.getPropertyValue("updbname");
        String table = tagInfo.getTagCode();

        String caseWhenSql = "";
        //当前计算的任务是有四级标签的三级标签，把sql查询的结果感觉映射规则，映射为四级标签名。不是，取计算结果即可
        if (rules.size() > 0){
            //有映射规则，就需要封装 case ... when
            String sql = rules.stream()
                                  .map(rule -> String.format(" when '%s' then '%s'  ", rule.getQueryValue(), rule.getSubTagValue()))
                                  .collect(Collectors.joining(" "));

            caseWhenSql = " case tagValue " +  sql + "  end tagValue ";

        }else{
            //没有四级标签的三级标签
            caseWhenSql = "tagValue";
        }

        //处理其他问题  把语句结尾的;去掉； 某些占位符doDate要替换为业务日期
        String sql = taskInfo.getTaskSql().replace(";", " ").replace("doDate", doDate);


        //格式化
        String insertSql = String.format(template, db, table, doDate, caseWhenSql, sql);

        log.error(insertSql);
        //System.out.println(insertSql);

        return insertSql;

    }

    /*
        确定当前task计算的结果表的建表语句


    关键: 是否是外部表。
            无所谓
         是否是分区表，按照什么分区
            是分区表，按照业务日期分区。随着数仓中数据的更新而更新。

    无关紧要:  store as  text|orc|parquet

    create [external] if not exists db.table(

    )
    comment '表注释'
    partition by xxx
    store as  text|orc|parquet
    location xxxx

    以计算性别标签为例

    create table if not exists upp220828.TAG_POPULATION_ATTRIBUTES_NATURE_GENDER(
    uid string,
    tagValue string
     )
    comment '性别'
    partitioned by (dt string)
    location 'hdfs://hadoop102:9820/upp220828/TAG_POPULATION_ATTRIBUTES_NATURE_GENDER'


     */
    private static String createTableSql(TagInfo tagInfo){

        String template = "create table if not exists %s.%s( uid string,  tagValue %s ) comment '%s'" +
                         " partitioned by (dt string)  location '%s' ";

        //根据当前计算的task确定变量部分，填充占位符
        String db = UPPUtil.getPropertyValue("updbname");
        String table = tagInfo.getTagCode();

        //根据查询的元数据中的tag_value_type确定 计算的结果值的类型
        String tagValueType = "";
        switch (tagInfo.getTagValueType()){
            case TagValueTypeConstant.TAG_VALUE_TYPE_LONG:  tagValueType = "bigint"; break;
            case TagValueTypeConstant.TAG_VALUE_TYPE_DECIMAL:  tagValueType = "decimal(16,2)"; break;
            case TagValueTypeConstant.TAG_VALUE_TYPE_STRING:  tagValueType = "string"; break;
            case TagValueTypeConstant.TAG_VALUE_TYPE_DATE:  tagValueType = "string"; break;
        }

        String location = UPPUtil.getPropertyValue("hdfsPath") + "/" + db + "/" + table.toLowerCase();

        //格式化，填充占位符
        String createTableSql = String.format(template, db, table, tagValueType, tagInfo.getTagName(), location);

        //System.out.println(createTableSql);
        log.error(createTableSql);

        return createTableSql;

    }

}
